package com.hdu.dwh.service.impl;

import java.sql.SQLException;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;

import com.hdu.dwh.utils.OperatorUtil;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hdu.dwh.dao.FlinkBaseDao;
import com.hdu.dwh.dao.FlinkTableDao;
import com.hdu.dwh.dao.PaimonBaseDao;
import com.hdu.dwh.dtos.model_dtos.BuildModelDto;
import com.hdu.dwh.dtos.model_dtos.ModelCreateDto;
import com.hdu.dwh.dtos.model_dtos.PreviewModelDto;
import com.hdu.dwh.mapper.ModelMapper;
import com.hdu.dwh.pojos.FlinkJob;
import com.hdu.dwh.pojos.Model;
import com.hdu.dwh.pojos.QueryResult;
import com.hdu.dwh.pojos.SQLBuilder.SQLInsertBuilder;
import com.hdu.dwh.pojos.SQLBuilder.SQLSelectBuilder;
import com.hdu.dwh.pojos.model_pojos.ConnectionColumn;
import com.hdu.dwh.pojos.model_pojos.MetaColumn;
import com.hdu.dwh.pojos.model_pojos.MetaTable;
import com.hdu.dwh.pojos.source_access_pojos.DataWarehouseDatabase;
import com.hdu.dwh.pojos.source_access_pojos.DataWarehouseTable;
import com.hdu.dwh.pojos.source_access_pojos.IColumn;
import com.hdu.dwh.service.*;
import com.hdu.dwh.service.flink_service.FlinkWebRestService;
import com.hdu.dwh.service.model_service.ConnectionColumnService;
import com.hdu.dwh.service.model_service.MetaColumnService;
import com.hdu.dwh.service.model_service.MetaTableService;
import com.hdu.dwh.vos.ModelVo;
import com.hdu.dwh.vos.model_vos.ConnectionColumnVo;
import com.hdu.dwh.vos.model_vos.MetaColumnVo;
import com.hdu.dwh.vos.model_vos.MetaTableVo;

import io.trino.jdbc.TrinoResultSet;
import lombok.extern.slf4j.Slf4j;

@Service
@Slf4j
public class ModelServiceImpl extends ServiceImpl<ModelMapper, Model> implements ModelService {

    private final FlinkWebRestService flinkWebRestService;

    @Value("${dwh.flink.jars.sql-jar.entry-class}")
    private String entryClass;

    @Value("${dwh.flink.jars.sql-jar.jar-name}")
    private String jarName;

    @Autowired
    private DataWarehouseTableService dwhTableService;

    @Autowired
    private MetaTableService metaTableService;

    @Autowired
    private MetaColumnService metaColumnService;

    @Autowired
    private ConnectionColumnService connectionColumnService;

    @Autowired
    private DataWarehouseDatabaseService dwhDatabaseService;

    @Autowired
    private PaimonBaseDao paimonBaseDao;

    @Autowired
    private TableRelationService tableRelationService;

    @Autowired
    private FlinkTableDao flinkTableDao;

    private final FlinkBaseDao flinkBaseDao;

    @Autowired
    private TrinoQueryService trinoQueryService;

    ModelServiceImpl(FlinkBaseDao flinkBaseDao, FlinkWebRestService flinkWebRestService) {
        this.flinkBaseDao = flinkBaseDao;
        this.flinkWebRestService = flinkWebRestService;
    }

    /**
     * 获取模型视图分页数据
     * 
     * @param page 页码
     * @param size 每页大小
     * @return 模型视图分页数据
     */
    @Override
    public Page<ModelVo> queryModelVoPage(int page, int size) {
        return this.baseMapper.queryModelVoPage(new Page<>(page, size));
    }

    /**
     * 获取模型视图列表
     * 
     * @return 模型视图列表
     */
    @Override
    public List<ModelVo> queryModelVoList() {
        return this.baseMapper.queryModelVoList();
    }

    /**
     * 根据ID获取模型视图
     * 
     * @param id 模型ID
     * @return 模型视图
     */
    @Override
    public ModelVo queryModelVoById(Integer id) {
        return this.baseMapper.queryModelVoById(id);
    }

    @Override
    @Transactional
    public ModelVo createModel(ModelCreateDto modelCreateDto) throws Exception {
        // TODO: 设置模型的所有者ID和项目ID
        // 创建湖仓表
        DataWarehouseTable modelTable = this.dwhTableService.createTable(modelCreateDto.getDatabaseId(),
                modelCreateDto.getName(), -1L, 2, modelCreateDto.getDescription());

        
        Model model = saveModel(modelCreateDto, modelTable);
        Long modelId = model.getId();

        // 1.1 创建模型的事实表
        MetaTable factTable = metaTableService.createMetaTable(modelCreateDto.getFactTable(), modelId, "fact");
        // 1.2 创建事实表的元数据列
        Set<MetaColumn> factColumns = metaColumnService.saveMetaColumns(modelCreateDto.getFactTable().getColumns(),
                factTable.getId());
        // 1.3 创建事实表和源表的关联关系
        DataWarehouseTable factSourceTable = this.dwhTableService
                .getById(modelCreateDto.getFactTable().getSourceTableId());
        this.tableRelationService.saveWithSourceTableIdAndTargetTableId(factSourceTable.getId(),
                model.getTableId());

        // 2.1 创建模型的维度表
        for (MetaTable dimensionTable : modelCreateDto.getDimensionTables()) {
            MetaTable savedDimensionTable = metaTableService.createMetaTable(dimensionTable, modelId, "dimension");
            Set<MetaColumn> dimensionColumns = metaColumnService.saveMetaColumns(dimensionTable.getColumns(),
                    savedDimensionTable.getId());
            DataWarehouseTable dimSourceTable = this.dwhTableService
                    .getById(dimensionTable.getSourceTableId());
            this.tableRelationService.saveWithSourceTableIdAndTargetTableId(dimSourceTable.getId(),
                    model.getTableId());
        }
        // 3.1 创建连接列
        connectionColumnService.createConnectionColumns(modelCreateDto.getJoinColumns(), modelId);
        // 4. 在paimon中创建表
        String databaseName = this.dwhDatabaseService.getById(modelCreateDto.getDatabaseId()).getDatabaseName();
        // 4.1 收集所有列
        Set<IColumn> columns = new HashSet<>();
        // 事实表的所有列
        columns.addAll(this.getColumnsFromMetaTable(modelCreateDto.getFactTable(), databaseName));
        // 维度表的所有列
        modelCreateDto.getDimensionTables().forEach(dimTable -> {
            columns.addAll(this.getColumnsFromMetaTable(dimTable, databaseName));
        });

        this.paimonBaseDao.createPaimonTable(
                databaseName,
                modelCreateDto.getName(),
                columns,
                modelCreateDto.getPrimaryKeys(),
                modelCreateDto.getPartitionKeys(),
                modelCreateDto.getOptions());
        return this.queryModelVoById(modelId.intValue());
    }

    private Model saveModel(ModelCreateDto modelCreateDto, DataWarehouseTable modelTable) {
        Model model = new Model();
        BeanUtils.copyProperties(modelCreateDto, model);
        model.setTableId(modelTable.getId());
        model.setCreatedBy(OperatorUtil.getCurrentOperator());
        this.save(model);
        return model;
    }

    private Set<IColumn> getColumnsFromMetaTable(MetaTable metaTable, String databaseName) {
        Set<IColumn> columns = new HashSet<>();
        for (MetaColumn column : metaTable.getColumns()) {
            if (column.getType() == null) {
                try {
                    column.setType(this.trinoQueryService.getColumnType(
                            databaseName, metaTable.getName(), column.getName()));
                } catch (Exception e) {
                    throw new RuntimeException("Failed to get column type for column: " + column.getName(), e);
                }
            }
            columns.add(new IColumn()
                    .setColumnName(column.getName())
                    .setColumnType(column.getType()));
        }
        return columns;
    }

    @Override
    public void updateModel(ModelCreateDto modelCreateDto, Long modelId) {
        // TODO 更新模型
        throw new UnsupportedOperationException("Unimplemented method 'updateModel'");
    }

    @Override
    // @Transactional
    public FlinkJob buildModel(BuildModelDto buildModelDto) throws Exception {
        Model model = this.getById(buildModelDto.getModelId());
        DataWarehouseTable modelTable = this.dwhTableService.getById(model.getTableId());
        DataWarehouseDatabase database = this.dwhDatabaseService.getById(modelTable.getDatabaseId());
        Set<MetaTable> dimensionTables = this.metaTableService.query()
                .eq("model_id", buildModelDto.getModelId())
                .eq("type", "dimension")
                .list()
                .stream()
                .collect(java.util.stream.Collectors.toSet());
        for (MetaTable dimTable : dimensionTables) {
            dimTable.setColumns(this.metaColumnService.query()
                    .eq("table_id", dimTable.getId())
                    .list()
                    .stream()
                    .collect(java.util.stream.Collectors.toSet()));
        }
        MetaTable factTable = this.metaTableService.query()
                .eq("model_id", buildModelDto.getModelId())
                .eq("type", "fact")
                .one();
        factTable.setColumns(this.metaColumnService.query()
                .eq("table_id", factTable.getId())
                .list()
                .stream()
                .collect(java.util.stream.Collectors.toSet()));
        Set<ConnectionColumn> connectionColumns = this.connectionColumnService.query()
                .eq("model_id", buildModelDto.getModelId())
                .list()
                .stream()
                .collect(java.util.stream.Collectors.toSet());

        // 选择出Columns
        LinkedHashSet<String> selectedColumns = new LinkedHashSet<>();
        selectedColumns.addAll(factTable.getColumns().stream().map(col -> {
            return col.getName();
        }).collect(java.util.stream.Collectors.toSet()));
        for (MetaTable dimTable : dimensionTables) {
            selectedColumns.addAll(dimTable.getColumns().stream().map(col -> {
               return col.getName();
            }).collect(java.util.stream.Collectors.toSet()));
        }

        SQLSelectBuilder selectBuilder = new SQLSelectBuilder();
        selectBuilder.select(selectedColumns)
                .from(database.getDatabaseName(), factTable.getName());
        connectionColumns.forEach(connCol -> {
            String joinType = connCol.getJoinType().toString();
            String leftTable = connCol.getLeftDatabaseName() + "." + connCol.getLeftTableName();
            String rightTable = connCol.getRightDatabaseName() + "." + connCol.getRightTableName();
            String leftColumn = connCol.getLeftColumnName();
            String rightColumn = connCol.getRightColumnName();
            selectBuilder.join(joinType, rightTable, leftTable + "." + leftColumn + " = " + rightTable + "." + rightColumn);
        });
        SQLInsertBuilder insertBuilder = new SQLInsertBuilder();
        insertBuilder.insertInto(database.getDatabaseName(), modelTable.getTableName())
                .setColumns(selectedColumns)
                .selectFrom(selectBuilder.build());
        String sql = insertBuilder.build();

        log.info("构建模型的SQL: {}", sql);
        return new FlinkJob().setJobid(this.paimonBaseDao.executeSQL(sql));
        // return this.sendSQLJobToFlink("SELECT 1");
    }

    @Override
    @Transactional
    public void deleteModel(Integer modelId) throws Exception {
        // 1. 删除模型相关的元数据
        ModelVo model = this.queryModelVoById(modelId);
        this.removeById(modelId);
        // 2. 删除模型相关的元数据
        Set<MetaTableVo> tables = model.getDimensionTables();
        tables.add(model.getFactTable());

        for (MetaTableVo metaTableVo : tables) {
            metaTableService.removeById(metaTableVo.getId());
            for (MetaColumnVo metaColumnVo : metaTableVo.getColumns()) {
                metaColumnService.removeById(metaColumnVo.getId());
            }
        }
        Set<ConnectionColumnVo> connectionColumns = model.getConnectionColumns();
        for (ConnectionColumnVo connectionColumnVo : connectionColumns) {
            connectionColumnService.removeById(connectionColumnVo.getId());
        }

        this.tableRelationService.query()
                .eq("target_table_id", model.getTableId())
                .list()
                .forEach(relation -> {
                    this.tableRelationService.removeById(relation.getId());
                });
        // 3. 删除模型对应的湖仓表
        DataWarehouseTable dwhTable = this.dwhTableService.getById(model.getTableId());
        // TODO: 如果模型没有构建，则无法删除湖仓表
        this.dwhTableService.deleteTable(dwhTable);
        // this.dwhTableService.removeById(model.getTableId());
    }

    @Override
    @Transactional
    public  FlinkJob createDimensionTable(DataWarehouseTable sourceTable, Set<IColumn> columns, String dimensionTableName, String description, Long databaseId) throws SQLException {
        DataWarehouseDatabase database = this.dwhDatabaseService.getById(sourceTable.getDatabaseId());
        DataWarehouseTable dimTable = new DataWarehouseTable()
                .setDatabaseId(database.getId())
                .setTableName(dimensionTableName)
                .setSourceTableId(sourceTable.getId())
                .setSourceTableName(sourceTable.getTableName())
                .setHierarchy(4)
                .setCreatedBy(OperatorUtil.getCurrentOperator())
                .setDatabaseId(databaseId)
                .setDescription(description);
        this.dwhTableService.save(dimTable);
        this.tableRelationService.saveWithSourceTableIdAndTargetTableId(sourceTable.getId(), dimTable.getId());
        StringBuilder selectFields = new StringBuilder();
        int index = 0;
        for (IColumn column : columns) {
            if (index > 0) {
                selectFields.append(", ");
            }
            selectFields.append(column.getColumnName());
            index++;
        }
        String jobId = this.paimonBaseDao.executeSQL(
                String.format("CREATE TABLE IF NOT EXISTS %s.%s AS SELECT %s FROM %s.%s",
                        database.getDatabaseName(), dimensionTableName,
                        selectFields.toString(),
                        database.getDatabaseName(), sourceTable.getSourceTableName()));
        return new FlinkJob().setJobid(jobId);

    }

    @Override
    public QueryResult previewModel(PreviewModelDto previewModel) throws Exception {
        String sql = this.generateSQL(previewModel.getFactTable(), previewModel.getDimensionTables(),
                previewModel.getJoinColumns());
        return this.trinoQueryService.executeQuery(sql);

    }

    private String generateSQL(MetaTable factTable, Set<MetaTable> tables, Set<ConnectionColumn> connectionColumns) {
        StringBuilder sqlBuilder = new StringBuilder();
        sqlBuilder.append("SELECT ");
        int colIndex = 0;
        for (MetaColumn column : factTable.getColumns()) {
            if (colIndex > 0) {
                sqlBuilder.append(", ");
            }
            sqlBuilder.append(column.getName());
            colIndex++;
        }
        for (MetaTable table : tables) {
            for (MetaColumn column : table.getColumns()) {
                if (colIndex > 0) {
                    sqlBuilder.append(", ");
                }
                sqlBuilder.append(column.getName());
                colIndex++;
            }
        }
        sqlBuilder.append(" FROM ");
        DataWarehouseTable sourceTable = this.dwhTableService.getById(factTable.getSourceTableId());
        DataWarehouseDatabase sourceDatabase = this.dwhDatabaseService.getById(sourceTable.getDatabaseId());
        sqlBuilder.append(sourceDatabase.getDatabaseName() + "." + sourceTable.getTableName());

        // 4.1 添加JOIN连接
        int dimIndex = 0;
        for (ConnectionColumn connectionColumnVo : connectionColumns) {
            if (dimIndex > 0) {
                sqlBuilder.append(" AND ");
            }

            String leftTableName = connectionColumnVo.getLeftTableName();
            String rightTableName = connectionColumnVo.getRightTableName();
            String leftColumnName = connectionColumnVo.getLeftColumnName();
            String rightColumnName = connectionColumnVo.getRightColumnName();
            String joinType = connectionColumnVo.getJoinType().toString();
            String databaseName = connectionColumnVo.getRightDatabaseName();
            sqlBuilder.append(String.format(" %s JOIN %s.%s ON %s.%s = %s.%s ", joinType.toUpperCase(), databaseName,
                    rightTableName,
                    leftTableName, leftColumnName, rightTableName, rightColumnName));
        }
        sqlBuilder.append("LIMIT 100");
        String sql = sqlBuilder.toString();
        log.info("生成的SQL: {}", sql);

        return sqlBuilder.toString();
    }

    @Override
    public QueryResult previewDimTable(MetaTable metaTable) throws Exception {
        DataWarehouseTable sourceTable = this.dwhTableService.getById(metaTable.getSourceTableId());
        StringBuilder selectFields = new StringBuilder();
        int index = 0;
        for (MetaColumn column : metaTable.getColumns()) {
            if (index > 0) {
                selectFields.append(", ");
            }
            selectFields.append(column.getName());
            index++;
        }
        DataWarehouseDatabase database = this.dwhDatabaseService.getById(sourceTable.getDatabaseId());
        String sql = String.format("SELECT %s FROM %s.%s LIMIT 100",
                selectFields.toString(),
                database.getDatabaseName(), sourceTable.getSourceTableName());
        return this.trinoQueryService.executeQuery(sql);
    }

    @Override
    public void deleteModelByTableId(Long tableId) {
        Model model = this.lambdaQuery().eq(Model::getTableId, tableId).one();
        if (model != null) {
            try {
                this.deleteModel(model.getId().intValue());
            } catch (Exception e) {
                throw new RuntimeException("删除模型失败: " + e.getMessage());
            }
        } else {
            throw new IllegalArgumentException("tableId无效，未找到对应的模型");
        }
    }
}
